#include	"builtin_libs.h"
#include	"users_libs.h"
statistics performance_vector[2000]; // this variable stores the statstics of LCR for each 1 second and finally saves them on disk.
//------------------------------------------------------------------------------
long getMiliSeconds() {
	//-------------------
	long time;
	//-------------------
	gettimeofday(&mtv, NULL);
	time = 1000 * mtv.tv_sec + mtv.tv_usec / 1000;
	return time;
}
//------------------------------------------------------------------------------
void utoBroadcast(int fd, short event, void * arg) {
	//-------------------
	int i, interval_msg_sent = 0;
	msg *vp;
	//-------------------
	ENTRANCE++;
	while (total_msg_sent < burstSize && interval_msg_sent < frequency) {
		vC[PROCESS_ID]++;
		msgsCounter++;
		msg *new_msg = (msg*) malloc(sizeof(msg));
		new_msg->type = 'm';
		new_msg->p_id = PROCESS_ID;

		for (i = 0; i < N; i++)
			new_msg->vC[i] = vC[i];

		new_msg->status = false;
		new_msg->startMiliSeconds = (double) getMiliSeconds();
		vp = (msg*) (rbsearch(new_msg, myTree));
		if (vp == NULL) {
			fprintf(stdout, "ERROR!!	Inside utoBroadcast method.	Tree Insertion Failed!	\n");
			exit(1);
		}
		if (send(successor_sid, new_msg, (sizeof(msg)), MSG_WAITALL) < 0) {
			perror("ERROR!! Inside utoBroadcast method. Sending failed!!");
			exit(1);
		}
		total_msg_sent++;
		interval_msg_sent++;
	}
	evtimer_add(timeout1, &tv1);
}
//------------------------------------------------------------------------------
void tryDeliver() {
	//-------------------
	void *vp;
	//-------------------
	head = (msg*) (rbmin(myTree));

	while (head != NULL && head->status) {
		deliveryCount++;
		avg_latency += getMiliSeconds() - head->startMiliSeconds;
		vp = (msg*) rbdelete(head, myTree);
		head = (msg*) (rbmin(myTree));
		free(vp);

		if (deliveryCount == expectedMessages)
			finalize_performance();
	}
}
//------------------------------------------------------------------------------
void message_processor(int fd, short event, void * arg) {
	//-------------------
	int i, recv_err, send_err, predecessor = 0, pre_predecessor = 0;
	msg *vp;
	char t[1];
	bool mt = false, at = false;
	//-------------------
	memset(m, '\0', sizeof(msg));
	memset(ak, '\0', sizeof(ack));

	if (terminated == 0) {
		recv_err = recv(fd, t, sizeof(char), MSG_PEEK);
		if (t[0] == 'm') {
			recv_err = recv(fd, m, sizeof(msg), MSG_WAITALL);
			mt = true;
		}
		if (t[0] == 'a') {
			recv_err = recv(fd, ak, sizeof(ack), MSG_WAITALL);
			at = true;
		}
		if (recv_err == 0) {
			perror("ERROR!!	Inside message_processor method. There was nothing to receive!  ");
			exit(1);
		}
		if (recv_err == sizeof(msg) || recv_err == sizeof(ack)) {
			if (mt) {
				if (m->p_id > 0)
					predecessor = m->p_id - 1;
				else
					predecessor = N - 1;

				if (m->p_id == 1)
					pre_predecessor = N - 1;
				else
					pre_predecessor = predecessor - 1;
			}
			if (at) {
				if (ak->p_id > 0)
					predecessor = ak->p_id - 1;
				else
					predecessor = N - 1;

				if (ak->p_id == 1)
					pre_predecessor = N - 1;
				else
					pre_predecessor = predecessor - 1;
			}

			if (t[0] == 'm') {
				if ((m->vC[m->p_id]) > (vC[m->p_id])) {
					msg *new_msg = (msg*) malloc(sizeof(msg));
					if (PROCESS_ID != predecessor) {
						msgsCounter++;
						memcpy(new_msg, m, sizeof(msg));

						int l = 0;
						for (l = 0; l < N; l++)
							new_msg->vC[l] = m->vC[l];

						if (send(successor_sid, m, sizeof(msg), MSG_WAITALL) < 0) {
							perror("ERROR!!	Inside message_processor method. Forwarding Failed! ");
							exit(1);
						}
						vp = (msg*) (rbsearch(new_msg, myTree));
						if (vp == NULL) {
							perror("ERROR! Inside message_processor method. Searching message of type m (1) in Tree failed.!  ");
							exit(1);
						}
					} else {
						msgsCounter++;
						memcpy(new_msg, m, sizeof(msg));

						int l = 0;
						for (l = 0; l < N; l++)
							new_msg->vC[l] = m->vC[l];

						vp = (msg*) (rbsearch(new_msg, myTree));
						if (vp == NULL) {
							perror("ERROR! Inside message_processor method. Searching message of type m (2) in Tree failed. ");
							exit(1);
						}
						vp->status = true;
						ak->type = 'a';
						ak->p_id = new_msg->p_id;
						for (i = 0; i < N; i++)
							ak->vC[i] = new_msg->vC[i];
						ak->status = false;
						send_err = send(successor_sid, ak, sizeof(ack), MSG_WAITALL);
						if (send_err < 0) {
							perror("ERROR! Inside message_processor method. Sending ACK failed!");
							exit(1);
						}
						tryDeliver();
					}
					updateClock(vC, m->vC);
				}
			}
			if (t[0] == 'a') {
				m->p_id = ak->p_id;
				m->type = 'm';

				for (i = 0; i < N; i++)
					m->vC[i] = ak->vC[i];
				m->status = ak->status;
				vp = (msg*) (rbfind(m, myTree));
				if (vp != NULL)
					vp->status = true;
				else {
					fprintf(stdout, "ERROR!!	Inside message_processor method. corresponding message for the received ACK in the tree not found! ");
					exit(1);
				}

				if (PROCESS_ID != pre_predecessor) {
					if (send(successor_sid, ak, sizeof(ack), MSG_WAITALL) < 0) {
						perror("ERROR!! Inside message_processor sending ACK failed!");
						exit(1);
					}
					tryDeliver();
				} else
					tryDeliver();
			}
		}
	}
}
//------------------------------------------------------------------------------
